package nr.nodehost;

import lombok.Data;

import java.io.*;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.file.CopyOption;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Arrays;

@Data
public class DataFrame {
    public static final int HEADER_LENGTH = 224;//not include length
    private int length;
    private int sourceHost;
    private int targetHost;
    private int frameNo;
    private int schemaId;
    private String sourceComponentId;
    private String sourceComponentPort;
    private String targetComponentId;
    private String targetComponentPort;
    private int rows;
    private int cols;
    private String dataFilename;
    private byte[] data;


    public DataFrame() {

    }
    public static int sRead(InputStream is, byte[] bufIO) throws IOException {
        int totalBytesRead = 0;
        int bufSize = bufIO.length;  // bufIO数组分配的空间就表示要读的数据长度
        int bytesRead;
        while (totalBytesRead < bufSize && (bytesRead = is.read(bufIO, totalBytesRead, bufSize - totalBytesRead)) != -1) {
            totalBytesRead += bytesRead;
        }
        return totalBytesRead;
    }
    public DataFrame(int frameNo, int schemaId, String sourceComponentId, String sourceComponentPort, String targetComponentId, String targetComponentPort, String dataFilename) {
        this.sourceHost = 0;
        this.targetHost = 0;
        this.frameNo = frameNo;
        this.schemaId = schemaId;
        this.sourceComponentId = sourceComponentId;
        this.sourceComponentPort = sourceComponentPort;
        this.targetComponentId = targetComponentId;
        this.targetComponentPort = targetComponentPort;
        this.dataFilename = dataFilename;
        try {

            File dataFile = new File(dataFilename);
            FileInputStream fis = new FileInputStream(dataFile);
            byte[] dimension = new byte[8];
            fis.read(dimension);
            this.rows = Utils.bytes2int(dimension, 0);
            this.cols = Utils.bytes2int(dimension, 4);
            fis.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static DataFrame fromDataBytes(byte[] frameData, int length, String schemaRunDir) {
        DataFrame frame = new DataFrame();
        try {
            int offset = 0;
            frame.sourceHost = Utils.bytes2int(frameData, offset);
            offset += 4;
            frame.targetHost = Utils.bytes2int(frameData, offset);
            offset += 4;
            frame.frameNo = Utils.bytes2int(frameData, offset);
            offset += 4;
            frame.schemaId = Utils.bytes2int(frameData, offset);
            offset += 4;
            frame.sourceComponentId = new String(frameData, offset, 36);
            offset += 36;
            frame.sourceComponentPort = new String(frameData, offset, 64).trim();
            //System.out.println(frame.sourceComponentPort+":"+frame.sourceComponentPort.length());
            offset += 64;
            frame.targetComponentId = new String(frameData, offset, 36);
            offset += 36;
            frame.targetComponentPort = new String(frameData, offset, 64).trim();
            //System.out.println(frame.targetComponentPort+":"+frame.getTargetComponentPort().length());
            offset += 64;
            frame.rows = Utils.bytes2int(frameData, offset);
            offset += 4;
            frame.cols = Utils.bytes2int(frameData, offset);
            offset += 4;

            frame.length = length - offset;
            frame.dataFilename = String.format("%s/%d/%s/input/%s_%06d.dat", schemaRunDir, frame.schemaId, frame.targetComponentId, frame.targetComponentPort, frame.frameNo);
            //=schemaRunDir+"/"+targetComponentId+"/"+dataFilename;
            System.out.println("frame length:" + frame.length);
            FileOutputStream fos = new FileOutputStream(frame.dataFilename);
            fos.write(Utils.int2bytes(frame.rows));
            fos.write(Utils.int2bytes(frame.cols));
            fos.close();
        } catch (Exception e) {
            e.printStackTrace();
            frame = null;
        }
        return frame;
    }


    @Override
    public String toString() {
        return "DataFrame{" +
                "length=" + length +
                ", sourceHost=" + sourceHost +
                ", targetHost=" + targetHost +
                ", frameNo=" + frameNo +
                ", schemaId=" + schemaId +
                ", sourceComponentId='" + sourceComponentId + '\'' +
                ", sourceComponentPort='" + sourceComponentPort + '\'' +
                ", targetComponentId='" + targetComponentId + '\'' +
                ", targetComponentPort='" + targetComponentPort + '\'' +
                ", data=" + Arrays.toString(data) +
                '}';
    }

    public long getLength() {
        return length;
    }

    public int getFrameNo() {
        return frameNo;
    }

    public int getSchemaId() {
        return schemaId;
    }




    public boolean appendData(byte[] data, int offset, int length) {
        boolean result = true;
        try {
            FileOutputStream fos = new FileOutputStream(dataFilename, true);
            fos.write(data, offset, length);
            fos.close();
        } catch (Exception e) {
            e.printStackTrace();
            result = false;
        }
        return result;
    }

    public boolean saveToFile(String filename) {
        boolean result = true;
        if (!dataFilename.equals(filename)) {
            try {
                File source = new File(dataFilename);
                File target = new File(filename);

                Files.copy(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING);
            } catch (Exception e) {
                e.printStackTrace();
                result = false;
            }
        } else
            System.out.println("same file");
        return result;
    }

    public boolean deleteDataFile() {
        boolean result = false;
        try {
            result = Files.deleteIfExists(new File(dataFilename).toPath());
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;

    }
    // Serialization method
    public  byte[] serialize() throws  IOException{
        int pos = 0;
        byte[] data = new byte[4 + 4 + 40 + 16 + 40 + 16 + 4 + this.length];
        Utils.int2bytes(this.schemaId,data, pos);  pos += 4;  //appid
        Utils.int2bytes(this.frameNo, data, pos);  pos += 4;  //frameno
        byte[] sid = this.getSourceComponentId().getBytes();
        System.arraycopy(sid,0, data, pos, sid.length);       pos += 40;

        byte[] sport = this.getSourceComponentPort().getBytes();
        System.arraycopy(sport, 0, data, pos, sport.length);  pos += 16;

        byte[] tid = this.getTargetComponentId().getBytes();
        System.arraycopy(tid,0, data, pos, tid.length);       pos += 40;

        byte[] tport = this.getTargetComponentPort().getBytes();
        System.arraycopy(tport, 0, data, pos, tport.length);  pos += 16;
        Utils.int2bytes(this.length, data, pos);  pos += 4;
        if(this.length > 0)
            System.arraycopy(this.data, 0, data, pos, this.length);
        return data;
    }
    public static DataFrame readOneFrame(Socket socket) {
        DataFrame df = new DataFrame();
        try {
            InputStream is = socket.getInputStream();
            byte[] buf = new byte[4];
            is.read(buf);
            df.setSchemaId(ByteBuffer.wrap(buf).getInt());
            is.read(buf);
            df.setFrameNo(ByteBuffer.wrap(buf).getInt());
            buf = new byte[40];
            is.read(buf);
            df.setSourceComponentId(new String(buf).trim());
            buf = new byte[16];
            is.read(buf);
            df.setSourceComponentPort(new String(buf).trim());
            buf = new byte[40];
            is.read(buf);
            df.setTargetComponentId(new String(buf).trim());
            buf = new byte[16];
            is.read(buf);
            df.setTargetComponentPort(new String(buf).trim());
            buf = new byte[4];
            is.read(buf);
            df.setLength(ByteBuffer.wrap(buf).getInt());
            buf = new byte[(int) df.getLength()];
            is.read(buf);
            df.setData(buf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return df;
    }

    // Deserialization method
    public static DataFrame deserialize(byte[] bytes) throws IOException, ClassNotFoundException {
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        DataFrame frame = new DataFrame();
        // Read schemaId and frameNo as ints
        frame.schemaId = buffer.getInt();
        frame.frameNo  = buffer.getInt();

        // Read sourceComponentId as string (32 bytes)
        byte[] sourceComponentIdBytes = new byte[40];
        buffer.get(sourceComponentIdBytes);
        frame.sourceComponentId = new String(sourceComponentIdBytes).trim();

        // Read sourceComponentPort as string (16 bytes)
        byte[] sourceComponentPortBytes = new byte[16];
        buffer.get(sourceComponentPortBytes);
        frame.sourceComponentPort = new String(sourceComponentPortBytes).trim();
        // Read targetComponentId as string (32 bytes)
        byte[] targetComponentIdBytes = new byte[40];
        buffer.get(targetComponentIdBytes);
        frame.targetComponentId = new String(targetComponentIdBytes).trim();
        // Read targetComponentPort as string (16 bytes)
        byte[] targetComponentPortBytes = new byte[16];
        buffer.get(targetComponentPortBytes);
        frame.targetComponentPort = new String(targetComponentPortBytes).trim();

        // Read length as int
        frame.length = buffer.getInt();
        // Read data
        frame.data = new byte[(int) frame.length];
        buffer.get(frame.data);

        return frame;
    }

}
